package com.ctrip.ibu.member.broker.controller;


import com.alibaba.fastjson.JSONObject;
import com.ctrip.ibu.member.broker.module.BrokerInfo;
import com.ctrip.ibu.member.broker.module.BrokerNode;
import com.ctrip.ibu.member.broker.module.MessageInfo;
import com.ctrip.ibu.member.broker.module.RequestQueueInfo;
import com.ctrip.ibu.member.broker.module.TopicMessage;
import com.ctrip.ibu.member.common.result.HandleResult;
import com.ctrip.ibu.member.common.util.HttpClientUtil;
import com.google.common.collect.Lists;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
@Api(description ="nameserver")
@RestController
public class BrokerController {

    private static final Logger logger = LoggerFactory.getLogger(BrokerController.class);


    /**
     * 本地缓存， 存储消息队列
     */
    private static Map<String, LinkedBlockingQueue<MessageInfo>> queueMap = new ConcurrentHashMap<>();

    private static Map<String, String> topicSerialNumberMap = new ConcurrentHashMap<>();

    /**
     * nameserver ip列表
     * todo 后期改成读配置
     */
    private static final List<String> NAME_SERVER_IP_LIST = Lists.newArrayList("http://127.0.0.1:8081", "2.2.2.2:8081");

    private static final String REGISTER_BROKER = "/registerBroker";

    //启动时注册broker
    static {
        // todo 待修改
        for (String ip : NAME_SERVER_IP_LIST) {
            //调用注册接口
            try {
                BrokerInfo brokerInfo = new BrokerInfo();
                brokerInfo.setIp(InetAddress.getLocalHost().getHostAddress());
                JSONObject object = HttpClientUtil.postJson("http://127.0.0.1:8081" + REGISTER_BROKER, JSONObject.toJSONString(brokerInfo));
            } catch (Exception e) {
                logger.error("init register broker error", e);
                throw new RuntimeException(e);
            }
        }
    }
    @ApiOperation("测试broker")
    @RequestMapping(value = "/testBrokerController", method = RequestMethod.GET)
    public String testBrokerController() {
        return "SUCCESS!";
    }

    @ApiOperation("注册broker到nameserver")
    @RequestMapping(value = "/registerBroker", method = RequestMethod.GET)
    public String registerBroker() throws Exception {
        InetAddress addr = InetAddress.getLocalHost();
        BrokerInfo brokerInfo = new BrokerInfo();
        brokerInfo.setIp(addr.getHostAddress());
        JSONObject object = HttpClientUtil.postJson("http://127.0.0.1:8081" + REGISTER_BROKER, JSONObject.toJSONString(brokerInfo));
        return object.toString();
    }

    @ApiOperation("broker是否存活")
    @RequestMapping(value = "/isActive", method = RequestMethod.GET)
    public BrokerNode isActive() throws UnknownHostException {
        String hostAddress = InetAddress.getLocalHost().getHostAddress();
        return new BrokerNode(hostAddress, true);
    }


    /**
     * 获得消息接口，提供给消费者获得消息
     *
     * @param requestQueueInfo
     * @return
     * @throws InterruptedException
     */
    @ApiOperation("获取消息")
    @RequestMapping(value = "/getQueue", method = RequestMethod.POST)
    public MessageInfo getQueue(@RequestBody RequestQueueInfo requestQueueInfo) throws InterruptedException {
        MessageInfo messageInfo = new MessageInfo();
        messageInfo.setTopic(requestQueueInfo.getTopic());
        if (Objects.nonNull(requestQueueInfo.getTopic()) && Objects.nonNull(queueMap.get(requestQueueInfo.getTopic()))) {
            LinkedBlockingQueue<MessageInfo> queue = queueMap.get(requestQueueInfo.getTopic());
            if (!queue.isEmpty()) {
                //todo 收到消费成功响应才删除
                MessageInfo message = queue.peek();
                topicSerialNumberMap.put(requestQueueInfo.getTopic(), requestQueueInfo.getSerialNumber());
                messageInfo.setMessageJson(message.getMessageJson());
                messageInfo.setMessageId(message.getMessageId());
                messageInfo.setSupportThisTopic(true);
                logger.info("getQueue success! topic:" + requestQueueInfo.getTopic() + "messageInfo:", messageInfo.toString());
                return messageInfo;
            } else {
                messageInfo.setAllFinished(true);
            }
        } else {
            messageInfo.setSupportThisTopic(false);
        }
        logger.info("getQueue failed! topic:" + requestQueueInfo.getTopic() + "messageInfo:", messageInfo.toString());
        return messageInfo;
    }


    /**
     * 消费消息成功回调接口
     *
     * @param requestQueueInfo
     * @return
     * @throws InterruptedException
     */
    @ApiOperation("获取消息回调接口，确认消费成功后删除消息内容")
    @RequestMapping(value = "/getQueueCallback", method = RequestMethod.POST)
    public HandleResult getQueueCallback(@RequestBody RequestQueueInfo requestQueueInfo) throws InterruptedException {
        if (Objects.nonNull(topicSerialNumberMap.get(requestQueueInfo.getTopic()))) {
            String serialNumber = topicSerialNumberMap.get(requestQueueInfo.getTopic());
            if (Objects.nonNull(serialNumber) && serialNumber.equals(requestQueueInfo.getSerialNumber())) {
                //todo 收到消费成功响应才删除
                queueMap.get(requestQueueInfo.getTopic()).take();
                logger.info("getQueueCallback success! topic:" + requestQueueInfo.getTopic());
                return new HandleResult(true);
            }
        }
        logger.info("getQueueCallback failed! topic:" + requestQueueInfo.getTopic());

        return new HandleResult(false);
    }

    @ApiOperation("发送消息，存入消息")
    @RequestMapping(value = "/putQueue", method = RequestMethod.POST)
    public HandleResult putQueue(@RequestBody TopicMessage topicMessage) throws InterruptedException {
        if (topicMessage.getMessageList().isEmpty()) {
            return new HandleResult(false);
        }

        for (MessageInfo message : topicMessage.getMessageList()) {
            if (Objects.nonNull(queueMap.get(topicMessage.getTopic()))) {
                LinkedBlockingQueue<MessageInfo> queue = queueMap.get(topicMessage.getTopic());
                //去重
                if(!queue.contains(message)) {
                    queue.put(message);
                }
            } else {
                LinkedBlockingQueue<MessageInfo> queue = new LinkedBlockingQueue<>();
                queue.put(message);
                queueMap.put(topicMessage.getTopic(), queue);
            }
        }
        logger.info("putQueue success! topicMessage:", topicMessage.toString());
        return new HandleResult(true);
    }
}
